package com.kingstar.ceres.data.db;

import com.kingstar.ceres.data.bo.CsvDataConfig;
import com.kingstar.ceres.data.constant.FileImportConstants;
import com.kingstar.ceres.data.handler.ITaskLockHandler;
import com.kingstar.ceres.data.holder.InsertHolder;
import com.kingstar.ceres.data.holder.InsertHolderFactory;
import com.kingstar.ceres.data.properties.FileParseProperties;
import com.kingstar.ceres.data.strategy.csv.BlockedCsvChannel;
import com.kingstar.ceres.data.util.FormatUtils;
import com.kingstar.ceres.data.util.MessageFormatUtils;
import com.kingstar.ceres.data.util.SnowIdUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class OriginalCsvChannelImpl extends BlockedCsvChannel {

    private final DataSource dataSource;
    private final ITaskLockHandler taskLockHandler;

    @Resource
    FileParseProperties properties;

    @Resource
    InsertHolderFactory holderFactory;

    public OriginalCsvChannelImpl(DataSource dataSource, ITaskLockHandler taskLockHandler) {
        this.dataSource = dataSource;
        this.taskLockHandler = taskLockHandler;
    }

    @Override
    public void process(List<List<String>> payload, CsvDataConfig config) {
        log.info("csv原始数据单批准备入库;size:{},table:{},regex:{}",
                payload.size(),config.getMatchTable(),config.getRegex());
        if (CollectionUtils.isEmpty(payload)){
            log.warn("入库数据集不可为空");
            return;
        }
        StopWatch watch=new StopWatch();
        watch.start();
        Connection connection=null;
        Statement statement=null;
        String date="";
        //判断是否配置文件日期
        if(StringUtils.isNotBlank(config.getFileDate())){
            date= FormatUtils.parseBody(config.getFileDate());
        }
        String[] jdbcTypes=null;
        //判断是否配置列类型
        if(StringUtils.isNotBlank(config.getColTypes())){
            jdbcTypes = config.getColTypes().split(";");
        }
        //判断是否存在配置InsertHolder
        final InsertHolder holder = holderFactory.fetchHolder(config.getInsertHolder());
        Set<Integer> ignoreIdxes=new HashSet<>();
        //判断是否有某些列忽略不插入数据库
        if(StringUtils.isNotBlank(config.getIgnoreIdx())){
            ignoreIdxes=Arrays.stream(config.getIgnoreIdx().split(",")).map(Integer::parseInt)
                    .collect(Collectors.toSet());
        }
        StringBuilder sql = new StringBuilder("INSERT ALL ");
        try {
            connection = dataSource.getConnection();
            watch.split();
            connection.setAutoCommit(false);
            statement = connection.createStatement();
            List<Long> ids=null;
            if(holder!=null){
                ids=new ArrayList<>();
            }
            for (int i = 0; i < payload.size(); i++) {
                List<String> records=payload.get(i);
                if (holder!=null){
                   records=holder.preInsert(records);
                }
                String s;
                //如果未配置列类型则直接按表顺序插入
                if(jdbcTypes==null || jdbcTypes.length==0){
                    s=sqlGenerate(records,config.getGenerateId(),date);
                }else {
                    //如果配置列类型则直接按字段配置顺序插入
                    s=sqlGenerate(records,date,jdbcTypes,ignoreIdxes,ids);
                }
                sql.append("INTO ").append(config.getMatchTable());
                if(StringUtils.isNotBlank(config.getTableCols())){
                    sql.append(" (").append(config.getTableCols()).append(")");
                }
                sql.append(" VALUES (")
                .append(s)
                .append(") ");
                if ((i+1)%properties.getCsv().getInsertCommit()==0 || i+1==payload.size()){
                    sql.append("SELECT 1 from DUAL");
                    String str = sql.toString();
                    if (log.isDebugEnabled()){
                        log.debug("csv原始数据单批入库sql:{}",str);
                    }
                    statement.executeUpdate(str);
                    connection.commit();
                    if(holder!=null){
                        holder.afterInsert(config,ids);
                        ids.clear();
                    }
                    sql = new StringBuilder("INSERT ALL ");
                }
            }
            log.info("csv原始数据单批入库完成;size:{},table:{},单次插入耗时:{}ms,获取连接耗时:{}",
                    payload.size(),config.getMatchTable(),watch.getTime(),watch.getSplitTime());
            watch.stop();
        } catch (SQLException e) {
            String format = MessageFormatUtils.format("csv原始数据单批入库失败;size:{},table:{},err sql:{};",
                    payload.size(), config.getMatchTable(), sql);
            log.error(format,e);
            taskLockHandler.partFail(config.getTaskId(),config.getMatchTable(), config.getOpDate(), e.getMessage());
        }finally {
            payload.clear();
            close(null,statement,connection);
        }
    }

    private void close(ResultSet rs, Statement stmt, Connection conn){
        if(rs!=null)try{rs.close();}catch(Exception e){e.printStackTrace();}
        if(stmt!=null)try{stmt.close();}catch(Exception e){e.printStackTrace();}
        //这里的连接是返回到了池中
        if(conn!=null)try{conn.close();}catch(Exception e){e.printStackTrace();}
    }

    private String sqlGenerate(List<String> list,int generateId,String date){
        StringBuffer value;
        if(FileImportConstants.GENERATE_ID==generateId){
            value = new StringBuffer(SnowIdUtils.uniqueLong() +",'");
        }else {
            value = new StringBuffer("'");
        }
        //当字段本身包含单引号时需要特殊处理,不然sql拼接会错误
        for (String s : list) {
            value.append(s.contains("'")?s.replace("'","''"):s)
                .append("','");
        }
        //处理文件日期
        if(StringUtils.isNotBlank(date)){
            value.append(date+"',");
        }else {
            value.deleteCharAt(value.length()-1);
        }
        value.append("sysdate");
//        list.clear();不可清理内层list
        return value.toString();
    }

    private String sqlGenerate(List<String> list,String date,String[] jdbcTypes,Set<Integer> ignoreIdxes,
                               List<Long> ids){
        StringBuffer value = new StringBuffer();
        int i=0;
        for (String type:jdbcTypes){
            while (ignoreIdxes.contains(i)){
                i++;
            }
            //生成id
            if(FileImportConstants.GENERATE.equals(type)){
                Long id=SnowIdUtils.uniqueLong();
                if(ids!=null){
                    ids.add(id);
                }
                value.append(id +",");
                continue;
            }
            //系统函数
            if(type.startsWith(FileImportConstants.SYSTEM_FUNCTION)){
                value.append(type.substring(1)).append(",");
                continue;
            }
            //文件日期
            if(FileImportConstants.FILE_DATE.equals(type)){
                value.append("'").append(date).append("',");
                continue;
            }
            //字符型
            if(FileImportConstants.TYPE_STRING.equals(type)){
                String s=list.get(i++);
                value.append("'").append(s.contains("'")?s.replace("'","''"):s)
                        .append("',");
                continue;
            }
            //数字型
            if(FileImportConstants.TYPE_NUMBER.equals(type)){
                String s=list.get(i++);
                value.append(s).append(",");
                continue;
            }
            //占位符  TO_TIMESTAMP('{PLACE_HOLDER}', 'YYYY-MM-DD HH24:MI:SS.FF')   TO_DATE('{PLACE_HOLDER}', 'YYYY-MM-DD')
            if(type.contains(FileImportConstants.PLACE_HOLDER)){
                value.append(type.replace(FileImportConstants.PLACE_HOLDER,list.get(i++))).append(",");
            }
        }
        value.deleteCharAt(value.length()-1);
//        list.clear();不可清理内层list
        return value.toString();
    }

}
